package chapter4.eg5;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.junit.Before;
import org.junit.Test;

public class RecordOffsetApp {
  private KafkaConsumer<String, String> consumer;

  @Before
  public void before() {
    Properties properties = new Properties();
    properties.put("bootstrap.servers", "localhost:9092");
    properties.put("group.id", "CountryCounter");
    properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    consumer = new KafkaConsumer<>(properties);
    consumer.subscribe(Collections.singletonList("customerCountries"), new SaveOffsetsOnRebalance(consumer));
  }

  @Test
  public void testRecordOffset() {
    consumer.poll(Duration.ofMillis(0));
    for (TopicPartition partition : consumer.assignment()) {
      consumer.seek(partition, getOffsetFromDB(partition));
    }

    while (true) {
      ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
      for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
        storeRecordIntoDB(record);
        storeOffsetIntoDB(record);
      }
      commitDBTransaction();
    }
  }

  private void commitDBTransaction() {

  }

  private void storeOffsetIntoDB(ConsumerRecord<String, String> record) {

  }

  private void storeRecordIntoDB(ConsumerRecord<String, String> record) {

  }

  private void processRecord(ConsumerRecord<String, String> record) {

  }

  private long getOffsetFromDB(TopicPartition partition) {
    return -1;
  }

}
